package org.qy.star_server.grpc;


import com.google.protobuf.ByteString;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.qy.star_common.enums.ExchangeType;
import org.qy.star_grpc.core_api.BrokerCoreApiGrpc;
import org.qy.star_grpc.core_api.BrokerCoreApiProto;
import org.qy.star_grpc.core_api.BrokerCoreApiProto.*;
import org.qy.star_server.core.BrokerServer;
import org.qy.star_server.core.ConsumerEnv;
import org.qy.star_server.core.Queue;
import org.qy.star_server.core.Token;
import org.qy.star_server.data.center.DiskDataCenter;
import org.qy.star_server.data.center.MemoryDataCenter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author yinger
 * @description TODO
 * @date 2024/3/30 15:28
 **/
@Slf4j
public class BrokerService extends BrokerCoreApiGrpc.BrokerCoreApiImplBase {
    private final MemoryDataCenter memory = MemoryDataCenter.getMemoryBean();

    // 存放客户端的流 以便在需要验证客户端连接状态时获取到 从而向客户端发起验证请求
    @Data
    static class ConsumerHeart {
        String consumerTar; // 区分客户端
        String heartVersionUid;// uuid 由客户端生成的 区分上一次心跳
        AtomicBoolean isAlive = new AtomicBoolean(false);
        boolean online;
    }

    private static final ConcurrentHashMap<String, ConsumerHeart> heartMap = new ConcurrentHashMap<>();

    @Override
    public void userDeclare(BrokerCoreApiProto.User request, StreamObserver<BrokerCoreApiProto.UnifiedResponse> responseObserver) {
        // 1. 读取请求
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        // 2. 处理请求
        String msg = user.userDeclare();
        boolean success = msg.isEmpty();
        BrokerCoreApiProto.UnifiedResponse response = BrokerCoreApiProto.UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        // 3. 返回响应
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void userDelete(BrokerCoreApiProto.User request, StreamObserver<BrokerCoreApiProto.UnifiedResponse> responseObserver) {
        // 1. 读取请求
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        // 2. 处理请求
        String msg = user.userDelete();
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        // 3. 返回响应
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void userUpdate(User request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String newPassword = request.getNewPassword();
        String msg = user.userUpdate(newPassword);
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void virtualHostDeclare(VirtualHost request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String virtualHostName = request.getVirtualHostName();
        String msg = user.virtualHostDeclare(virtualHostName);
        System.out.println("客户端调用结果：" + msg);
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        // ！！！ 响应完毕后告grpc
        responseObserver.onCompleted();
    }

    @Override
    public void virtualHostDelete(VirtualHost request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String virtualHostName = request.getVirtualHostName();
        String msg = user.virtualHostDelete(virtualHostName);
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void exchangeDeclare(ExchangeDeclare request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        int exchangeType = request.getExchangeType();
        ExchangeType type = null;
        if (exchangeType == 0) {
            type = ExchangeType.DIRECT;
        } else if (exchangeType == 1) {
            type = ExchangeType.FANOUT;
        } else if (exchangeType == 2) {
            type = ExchangeType.TOPIC;
        }
        String msg = user.exchangeDeclare(request.getVirtualHostName(), request.getExchangeName(), type, request.getDurable(), request.getAutoDelete(), null);
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void exchangeDelete(ExchangeDelete request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String msg = user.exchangeDelete(request.getVirtualHostName(), request.getExchangeName());
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void queueDeclare(QueueDeclare request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String msg = user.queueDeclare(request.getVirtualHostName(), request.getQueueName(), request.getDurable(), request.getExclusive(), request.getAutoDelete(), null);
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void queueDelete(QueueDelete request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String msg = user.queueDelete(request.getVirtualHostName(), request.getQueueName());
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void bind(Bind request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String msg = user.queueBind(request.getVirtualHostName(), request.getExchangeName(), request.getQueueName(), request.getBindingKey());
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void unBind(UnBind request, StreamObserver<UnifiedResponse> responseObserver) {

        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String msg = user.queueUnbind(request.getVirtualHostName(), request.getExchangeName(), request.getQueueName());
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void publish(Publish request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String msg = user.basicPublish(request.getVirtualHostName(), request.getExchangeName(), request.getRoutingKey(), request.getMessageId(), request.getDeliveryMode(), request.getPayload().toByteArray());
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }


    @Override
    public StreamObserver<Subscribe> subscribe(StreamObserver<Message> responseObserver) {
        final ConsumerEnv[] consumerEnv = {null};
        return new StreamObserver<Subscribe>() {
            @Override
            public void onNext(Subscribe request) {
                Queue queue = memory.selectQueue(request.getUsername(), request.getVirtualHostName(), request.getQueueName());
                org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
                Token token = new Token(request.getUsername(), request.getVirtualHostName(), request.getQueueName());
                String msg = user.basicConsume(request.getVirtualHostName(), request.getConsumerTag(), token, request.getAutoAck());
                responseObserver.onNext(Message.newBuilder().setMessageId("".equals(msg) ? "1" : "0").build());
                consumerEnv[0] = new ConsumerEnv(
                        request.getConsumerTag(),
                        request.getUsername(),
                        request.getVirtualHostName(),
                        request.getQueueName(),
                        new LinkedBlockingQueue<>(),
                        responseObserver
                );
                consumerEnv[0].setOnline(true);
                consumerEnv[0].setAutoAck(request.getAutoAck());
                queue.addConsumerEnv(consumerEnv[0]);
            }

            @Override
            public void onError(Throwable t) {
                log.info("消费者{}已离线，结束消息推送 ... ", consumerEnv[0].getConsumerTag());
                Queue queue = memory.selectQueue(consumerEnv[0].getUsername(), consumerEnv[0].getVirtualHostName(), consumerEnv[0].getQueueName());
                queue.removeConsumerEnv(consumerEnv[0]);
                consumerEnv[0].setOnline(false);
            }

            @Override
            public void onCompleted() {
                log.info("消费者离线");
                log.info("消息推送结束！");
                Queue queue = memory.selectQueue(consumerEnv[0].getUsername(), consumerEnv[0].getVirtualHostName(), consumerEnv[0].getQueueName());
                queue.removeConsumerEnv(consumerEnv[0]);
                consumerEnv[0].setOnline(false);
            }
        };
    }

    @Override
    public void ack(Ack request, StreamObserver<UnifiedResponse> responseObserver) {
        org.qy.star_server.core.User user = new org.qy.star_server.core.User(request.getUsername(), request.getPassword());
        String msg = user.basicAck(request.getVirtualHostName(), request.getQueueName(), request.getMessageId());
        boolean success = msg.isEmpty();
        UnifiedResponse response = UnifiedResponse.newBuilder().setFailMsg(msg).setSuccess(success).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

}
